package com.kafka.comsumer.listener;

import com.rabbitmq.entity.User;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.HashMap;

/**
 * @作者: 林江
 * @创建时间: 2024/4/2
 * @功能:
 */
@Component
public class KafkaConsumerListener01 {

    //    @KafkaListener(id = "myId", topics = "test-topic")
    public void testTopic(String in) {
        System.out.println(in);
    }

    //    @KafkaListener(id = "one", topics = "one")
    public void one(String data) {
        System.out.println(data + "------------------------");
    }

    @KafkaListener(id = "tow", topics = "tow"
//            , errorHandler = "customErrorHandler" // 执行错误时的处理
//            , filter = "customFilter"
    )
    @SendTo // 响应
    public Message tow(String in, Consumer<?, ?> c) {
        System.out.println(in);
        MessageHeaders headers = new MessageHeaders(new HashMap<String, Object>() {{
//            put(KafkaHeaders.REPLY_TOPIC, "reply");
            put(KafkaHeaders.TOPIC, "reply");
//            put(KafkaHeaders.CORRELATION_ID);
        }});


//        MessageBuilder.withPayload("消费者成功接收到消息")
//                .setHeader(KafkaHeaders.TOPIC, replyTo)
//                .setHeader(KafkaHeaders.KEY, 42)
//                .setHeader(KafkaHeaders.CORRELATION_ID, correlation)
//                .build();
        return MessageBuilder.createMessage("消费者成功接收到消息", headers);
    }

    //    @Bean
    public KafkaListenerErrorHandler customErrorHandler() {
        return (m, e) -> {
            return m;
        };
    }

    //    @Bean
    public FilteringMessageListenerAdapter filteringMessageListenerAdapter() {
        // 需要拦截的监听器

        return new FilteringMessageListenerAdapter(null, consumerRecord -> {
            System.out.println(consumerRecord + " --> 消息被拦截之后，校验消息是否已经被消费过。");
            return false;
        });
    }

    @KafkaListener(id = "domain", topics = "domain")
    public void domain(User user) {
        System.out.println(user + "------------------------");
    }

    @KafkaListener(id = "one", topics = "one")
    public void one01(@Payload String data) {
        System.out.println(data + "------------------------");
    }
}
